package com.zim.access.infrastructurre.etcd;

import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.Lease;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.etcd.jetcd.options.PutOption;
import io.grpc.stub.CallStreamObserver;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.nio.charset.Charset;

@Slf4j
public class EtcdClientAdvancedImpl implements EtcdClientAdvanced {

    @Autowired
    private Client client;

    @Override
    public LeaseClient leaseClient() {
        return new LeaseClientImpl();
    }

    private ByteSequence bytesOf(String key) {
        return ByteSequence.from(key, Charset.defaultCharset());
    }

    class LeaseClientImpl implements LeaseClient {

        @Override
        public void keepAlive(String key, String value) {
            Lease leaseClient = client.getLeaseClient();

            leaseClient.grant(60)
                    .thenAccept(grantResponse -> {
                        long leaseId = grantResponse.getID();

                        if (log.isInfoEnabled()) {
                            log.info("[{}] apply lease success，leaseId is [{}]", key, Long.toHexString(leaseId));
                        }

                        client.getKVClient().put(
                                bytesOf(key),
                                bytesOf(value),
                                PutOption.newBuilder().withLeaseId(leaseId).build())
                                .thenAccept(putResponse -> {
                                    leaseClient.keepAlive(leaseId, new CallStreamObserver<LeaseKeepAliveResponse>() {
                                        @Override
                                        public boolean isReady() {
                                            return false;
                                        }

                                        @Override
                                        public void setOnReadyHandler(Runnable runnable) {

                                        }

                                        @Override
                                        public void disableAutoInboundFlowControl() {

                                        }

                                        @Override
                                        public void request(int i) {

                                        }

                                        @Override
                                        public void setMessageCompression(boolean b) {

                                        }

                                        @Override
                                        public void onNext(LeaseKeepAliveResponse leaseKeepAliveResponse) {
                                            if (log.isInfoEnabled()) {
                                                log.info("[{}] relet success, TTL[{}]", Long.toHexString(leaseId), leaseKeepAliveResponse.getTTL());
                                            }
                                        }

                                        @Override
                                        public void onError(Throwable t) {
                                            log.error("[{}] operation onError", Long.toHexString(leaseId), t);
                                        }

                                        @Override
                                        public void onCompleted() {
                                            log.info("[{}] operation onCompleted", Long.toHexString(leaseId));
                                        }
                                    });
                                });
                    });
        }
    }
}
